package com.xiaoshu.disruptor;

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.xiaoshu.common.Constant;
import com.xiaoshu.disruptor.handler.SaveScoreEventHandler;
import com.xiaoshu.disruptor.handler.SendMessageEventHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 功能说明： 使用 Disruptor 内存队列，缓存积分数据；并通过多个handler进行处理（职责链模式）
 * （1）SaveScoreEventHandler: 负责保存数据到数据库；
 * （2）SendMessageEventHandler：负责发送websocket 消息推送到前端页面；
 *
 *  com.xiaoshu.disruptor
 *
 * <p>
 * Original @Author: deane.jia-贾亮亮,2021/6/15 14:12
 * <p>
 * Copyright (C)2012-2021 小树盛凯科技 All rights reserved.
 */

@Component
@Slf4j
public class MessageDisruptorQueue {

    private static RingBuffer<MessageEvent> ringBuffer;

    private Executor executor = new ThreadPoolExecutor(Constant.THREADPOOL_CORE_POOL_SIZE, Constant.THREADPOOL_MAXI_NUM_POOL_SIZE,
            Constant.THREADPOOL_KEEP_ALIVE_TIME, TimeUnit.SECONDS, new LinkedBlockingQueue<>());

    private int bufferSize = 2 * 2048;

    private MessageEventFactory factory = new MessageEventFactory();

    @SuppressWarnings("deprecation")
    private Disruptor<MessageEvent> disruptor = new Disruptor<>(factory, bufferSize, executor);

    @SuppressWarnings("deprecation")
    MessageDisruptorQueue(@Autowired SaveScoreEventHandler saveScoreEventHandler,
                          @Autowired SendMessageEventHandler sendMessageMessageHandler) {
        log.info("start up the disruptor,  on date:{}", new Date().toLocaleString());
        disruptor.handleEventsWith(saveScoreEventHandler).then(sendMessageMessageHandler);
        ringBuffer = disruptor.getRingBuffer();
        disruptor.start();
    }

    /**
     * publish message into RingBuffer ;
     * @param methodInfoModel
     */
    public static void publishEvent(MethodInfoModel methodInfoModel) {
        Long sequence = ringBuffer.next();
        try {
            // get the entry in the Disruptor
            MessageEvent event = ringBuffer.get(sequence);
            // for the sequence, fill with data
            event.setMethodInfoModel(methodInfoModel);
        } finally {
            ringBuffer.publish(sequence);
        }
    }

}
